package com.essence.grabiec104.grab.executor;

import com.essence.common.model.BytesFrame;
import com.essence.grabiec104.grab.executor.handler.AbstractSocketFrameHandler;
import com.essence.grabiec104.grab.executor.handler.AnalysisHandler;
import com.essence.grabiec104.grab.executor.handler.YKFrameComparisonHandler;
import com.essence.grabiec104.grab.router.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.zeromq.SocketType;
import org.zeromq.ZMQ;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;

/**
 * @brief BytesFrame异步分发器
 * @details 将BytesFrame通过责任链模式异步分发给各个处理器, 每个处理器通过workerPair实现跨线程通信, 如果chainRoot为AbstractSocketFrameHandler类型，
 * 则需要添加workerPair, 每一个线程只创建一次workerPair
 */
@Component
public class AsyncHandlerDispatcher {
    private static final Logger LOG = LoggerFactory.getLogger(AsyncHandlerDispatcher.class);
    private ExecutorService executor;
    @Autowired
    private AnalysisHandler analysisHandler;
    @Autowired
    private YKFrameComparisonHandler ykFrameComparisonHandler;
    private final ThreadLocal<ZMQ.Socket> workerPairThreadLocal = new ThreadLocal<>();

    @PostConstruct
    public void init() {
        this.executor = SaveExecutor.addPool("grabber_dispatcher", 1);
        analysisHandler.setNext(ykFrameComparisonHandler);
    }

    /**
     * @brief 分发BytesFrame
     * @details 将BytesFrame通过责任链模式异步进行处理
     * @param frame 需要处理的BytesFrame
     */
    public void dispatch(BytesFrame frame) {
        // 为每一个线程创建一个workerPair，并且这个workerPair应该和该线程的生命周期一致，不需要显式关闭
        executor.submit(() -> {
            ZMQ.Socket workerPair = workerPairThreadLocal.get();
            try {
                if(workerPair == null) {
                    workerPair = Publisher.ZCtxParam.INSTANCE.getPublishCtx().createSocket(SocketType.PUSH);
                    workerPair.connect(Publisher.ZCtxParam.INSTANCE.getInprocName());
                    workerPairThreadLocal.set(workerPair);
                }
                analysisHandler.setWorkerSocket(workerPair);
                analysisHandler.handleFrame(frame);
            } catch (Exception ex) {
                LOG.error("dispatch error", ex);
            }
        });
    }
}
